package com.duan.laugh.gateway.Filters;

import com.duan.laugh.common.event.pojo.dto.EventDTO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.cloud.gateway.support.BodyInserterContext;
import org.springframework.core.Ordered;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.codec.HttpMessageReader;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.BodyInserter;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.server.HandlerStrategies;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.time.Clock;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;

import static com.duan.laugh.common.event.constants.EventConstant.*;
import static org.springframework.cloud.gateway.support.ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR;

/**
 * <p>
 * 公共Filter
 * </p>
 *
 * @author kylin
 * @since 2019-11-13
 **/
@Slf4j
@Component
public class CommonFilter implements GlobalFilter, Ordered {
    private static final String CACHE_REQUEST_BODY_OBJECT_KEY = "cachedRequestBodyObject";
    private static final String REQUEST_TIME = "requestTime";
    private final List<HttpMessageReader<?>> messageReaders;
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public CommonFilter(){
        this.messageReaders = HandlerStrategies.withDefaults().messageReaders();
    }

    @Override
    public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {

        exchange.getAttributes().put(REQUEST_TIME, System.currentTimeMillis());
//        ServerRequest serverRequest = ServerRequest.create(exchange,
//                this.messageReaders);
        MediaType mediaType = exchange.getRequest().getHeaders().getContentType();
        if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {
            ServerRequest serverRequest = ServerRequest.create(exchange, this.messageReaders);
//            ServerHttpRequest request = exchange.getRequest();
//            HttpHeaders headers = request.getHeaders();
//            MultiValueMap<String, HttpCookie> cookies = request.getCookies();
//            MultiValueMap<String, String> queryParams = request.getQueryParams();
//            Flux<DataBuffer> body = request.getBody();


//            serverRequest.
            // TODO: flux or mono
            Mono<String> modifiedBody = serverRequest.bodyToMono(String.class)
                    // .log("modify_request_mono", Level.INFO)
                    .flatMap(body -> {
//                        log.info("post请求内容为：{}",body);
                        exchange.getAttributes().put(CACHE_REQUEST_BODY_OBJECT_KEY, body);
                        return Mono.just(body);
                    });

            BodyInserter bodyInserter = BodyInserters.fromPublisher(modifiedBody, String.class);
            HttpHeaders headers = new HttpHeaders();
            headers.putAll(exchange.getRequest().getHeaders());

            // the new content type will be computed by bodyInserter
            // and then set in the request decorator
            headers.remove(HttpHeaders.CONTENT_LENGTH);

            // if the body is changing content types, set it here, to the bodyInserter
            // will know about it
//        if (config.getContentType() != null) {
//            headers.set(HttpHeaders.CONTENT_TYPE, config.getContentType());
//        }
            CachedBodyOutputMessage outputMessage = new CachedBodyOutputMessage(exchange,
                    headers);
            return bodyInserter.insert(outputMessage, new BodyInserterContext())
                    // .log("modify_request", Level.INFO)
                    .then(Mono.defer(() -> {
                        ServerHttpRequest decorator = decorate(exchange, headers,
                                outputMessage);
                        return chain.filter(exchange.mutate().request(decorator).build()).then(
                                Mono.fromRunnable(() -> {
                                    HttpStatus statusCode = exchange.getResponse().getStatusCode();
                                    if (statusCode == HttpStatus.OK){
                                        log.debug("请求响应成功，开始发送事件");
                                        doResponse(exchange);
                                    }
                                    log.debug("请求响应失败，状态码{}",statusCode);
                                }));
                    }));
        }
        return chain.filter(exchange).then(
                Mono.fromRunnable(() -> {
                    HttpStatus statusCode = exchange.getResponse().getStatusCode();
                    if (statusCode == HttpStatus.OK){
                        log.debug("请求响应成功，开始发送事件");
                        doResponse(exchange);
                    }
                    log.debug("请求响应失败，状态码{}",statusCode);
                }));
    }

    public void doResponse(ServerWebExchange exchange){
        Long startTime = exchange.getAttribute(REQUEST_TIME);
        URI uri = exchange.getAttribute(GATEWAY_REQUEST_URL_ATTR);
        ServerHttpRequest request = exchange.getRequest();
        HttpMethod method = request.getMethod();
        HttpHeaders headers = request.getHeaders();
        MediaType mediaType = exchange.getRequest().getHeaders().getContentType();
        if(uri !=null){
            String path = uri.getPath();
            String query = uri.getQuery();
            EventDTO eventDTO = new EventDTO();
            HashMap<String, String> eventData = new HashMap<>(1>>>3);

            List<String> authorization = headers.get(REQUEST_HEADER_AUTHORIZATION);
            // 登录获取token
            if(authorization != null && authorization.size()> 0){
                eventData.put(EVENT_DATA_TOKEN, authorization.get(0));
            }
            // json请求获取body
            if(MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)){
                eventData.put(REQUEST_BODY,exchange.getAttribute(CACHE_REQUEST_BODY_OBJECT_KEY));
            }
            eventData.put(REQUEST_PATH,path);
            eventData.put(REQUEST_METHOD,method.name());
            eventData.put(REQUEST_PARAM,query);


            eventDTO.setCreateTime(LocalDateTime.now()).setEventData(eventData).setType(EVENT_TYPE_URL);
            log.debug("发送事件到MQ开始");
            long start = System.currentTimeMillis();
            try{
                rabbitTemplate.convertAndSend(MQ_QUEUE_EVENT,eventDTO);
                log.debug("发送事件到MQ结束，耗时{}ms",System.currentTimeMillis() - start);
            }catch(Exception e){
                log.error("访问MQ出错,耗时："+ (System.currentTimeMillis() - start) +"ms",e);
            }

        }
        if (startTime != null) {
            log.info(exchange.getRequest().getURI().getRawPath() + ": " + (System.currentTimeMillis() - startTime) + "ms");
        }
    }


    ServerHttpRequestDecorator decorate(ServerWebExchange exchange, HttpHeaders headers,
            CachedBodyOutputMessage outputMessage) {
        return new ServerHttpRequestDecorator(exchange.getRequest()) {
            @Override
            public HttpHeaders getHeaders() {
                long contentLength = headers.getContentLength();
                HttpHeaders httpHeaders = new HttpHeaders();
                httpHeaders.putAll(super.getHeaders());
                if (contentLength > 0) {
                    httpHeaders.setContentLength(contentLength);
                }
                else {
                    // TODO: this causes a 'HTTP/1.1 411 Length Required' // on
                    // httpbin.org
                    httpHeaders.set(HttpHeaders.TRANSFER_ENCODING, "chunked");
                }
                return httpHeaders;
            }

            @Override
            public Flux<DataBuffer> getBody() {
                return outputMessage.getBody();
            }
        };
    }

    @Override
    public int getOrder() {
        return -1000;
    }

}
